package com.atguigu.tingshu.search.service.impl;
import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch._types.SortOrder;
import co.elastic.clients.elasticsearch.core.SearchResponse;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.nacos.client.naming.utils.CollectionUtils;
import com.atguigu.tingshu.common.util.PinYinUtils;
import com.atguigu.tingshu.model.album.BaseCategory1;
import com.atguigu.tingshu.model.album.BaseCategory3;
import com.atguigu.tingshu.search.repository.SuggestIndexRepository;
import com.atguigu.tingshu.vo.search.AlbumInfoIndexVo;
import org.springframework.data.elasticsearch.core.suggest.Completion;
import com.atguigu.tingshu.album.client.AlbumInfoFeignClient;
import com.atguigu.tingshu.common.execption.GuiguException;
import com.atguigu.tingshu.common.result.Result;
import com.atguigu.tingshu.model.album.AlbumInfo;
import com.atguigu.tingshu.model.album.BaseCategoryView;
import com.atguigu.tingshu.model.search.AttributeValueIndex;
import com.atguigu.tingshu.model.search.SuggestIndex;
import com.atguigu.tingshu.user.client.UserInfoFeignClient;
import com.atguigu.tingshu.vo.album.AlbumStatVo;
import com.atguigu.tingshu.vo.user.UserInfoVo;

import java.util.*;
import java.util.concurrent.*;
import java.util.stream.Collectors;

import com.atguigu.tingshu.model.search.AlbumInfoIndex;
import com.atguigu.tingshu.search.repository.AlbumInfoRepository;
import com.atguigu.tingshu.search.service.ItemService;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;

@Slf4j
@Service
@SuppressWarnings({"unchecked", "rawtypes"})
public class ItemServiceImpl implements ItemService {

    @Autowired
    private AlbumInfoRepository albumInfoRepository;

    @Autowired
    private AlbumInfoFeignClient albumInfoFeignClient;

    @Autowired
    private SuggestIndexRepository suggestIndexRepository;

    @Autowired
    private UserInfoFeignClient userInfoFeignClient;

    @Autowired
    private ThreadPoolExecutor threadPoolExecutor;

    @Autowired
    private ElasticsearchClient elasticsearchClient;


    @Autowired
    private StringRedisTemplate redisTemplate;

      static ExecutorService executorService = null;

    static {
        //快速得到一个线程池的api
         executorService = Executors.newFixedThreadPool(4);
    }


    /**
     * 异步的作用:一定能提高接口的响应速度.(异步的本质:压榨cpu)
     * 异步和快没有直接关系(异步涉及到多线程,实现多线程需要cpu频繁的切换线程(这一过程需要时间))  通常来说异步可以提高响应速度
     * 线程池分析: 第一次容器创建的时候,容器中是没有线程的,tomact需要船舰线程,后面调用,tomcat可以直接使用线程池里已经有的线程.
     * Feign和Nacos:底层有缓存机制,第一次需要网络传输,获取数据,但是后面的每一次都直接从本地缓存中直接获取第一次远程调用的结果(最主要),不需要网络传输
     * 这是为什么第一次调用慢,后面调用快的原因
     * @param albumId
     * @return
     */

    /**
     * 异步编排
     * 1)new Thread();
     * 2)创建一个线程池
     * TODO: 异步编排
     * @param albumId
     * @return
     */

    /**
     * 向线程池中提交任务有两种方式
     * 1.submit();
     * 参数类型:Runnable和Callable
     * 返回值类型:FutureTask<Void>和FutureTask<T>
     * 2.execute();
     * 参数类型:Runnable
     * 返回值类型:void
     *
     * 线程池与CountDownLatch是标准的写法搭配
     * @param albumId
     * @return
     */

    @Override
    @SneakyThrows
    public boolean albumOnSale(Long albumId) {

        //注意,CountDownLatch只能放在方法内部,不能放放到类中作为成员变量
        CountDownLatch countDownLatch = new CountDownLatch(4);

        //1.创建一个文档对象
        AlbumInfoIndex albumInfoIndex = new AlbumInfoIndex();
        long startTime = System.currentTimeMillis();
        //2.给文档对象赋值
        albumInfoIndex.setId(albumId);
        Future<Long> submit = executorService.submit(new Callable<Long>() {
            @Override
            public Long call() throws Exception {
                //2.1 远程调用专辑微服务获取专辑基本信息
                Result<AlbumInfo> albumInfoResult = albumInfoFeignClient.getAlbumInfoAndTag(albumId);
                AlbumInfo albumInfoData = albumInfoResult.getData();
                if (albumInfoData == null) {
                    log.error("搜索微服务远程调用专辑微服务失败");
                    throw new GuiguException(201, "搜索微服务远程调用专辑微服务失败");
                }
                albumInfoIndex.setAlbumTitle(albumInfoData.getAlbumTitle());
                albumInfoIndex.setAlbumIntro(albumInfoData.getAlbumIntro());
                albumInfoIndex.setCoverUrl(albumInfoData.getCoverUrl());
                albumInfoIndex.setIncludeTrackCount(albumInfoData.getIncludeTrackCount());
                albumInfoIndex.setIsFinished(albumInfoData.getIsFinished().toString());
                albumInfoIndex.setPayType(albumInfoData.getPayType());
                List<AttributeValueIndex> attributeValues = albumInfoData.getAlbumAttributeValueVoList().stream().map(albumAttributeValue -> {
                    AttributeValueIndex attributeValueIndex = new AttributeValueIndex();
                    attributeValueIndex.setAttributeId(albumAttributeValue.getAttributeId());
                    attributeValueIndex.setValueId(albumAttributeValue.getValueId());
                    return attributeValueIndex;
                }).collect(Collectors.toList());
                albumInfoIndex.setAttributeValueIndexList(attributeValues);
                countDownLatch.countDown(); //每执行完一个任务,就将countDownLatch减1,3
                return albumInfoData.getUserId();
            }
        });

        executorService.execute(new Runnable() {
            @Override
            public void run() {
                try {
                    Long userId = submit.get(); //获取到上一个任务的返回值(call方法的返回值) 且get方法会阻塞当前线程,直到上一个任务执行完毕,才会执行当前任务
                    //2.2 远程调用,用户微服务获取用户基本信息
                    Result<UserInfoVo> userInfoVoResult=userInfoFeignClient.getUserInfoByUserId(userId);
                    UserInfoVo userInfoVoData = userInfoVoResult.getData();
                    if (userInfoVoData == null){
                        log.error("搜索微服务远程调用用户微服务失败,获取用户信息失败");
                        throw new GuiguException(201,"搜索微服务远程调用用户微服务失败,获取用户信息失败");
                    }
                    albumInfoIndex.setAnnouncerName(userInfoVoData.getNickname());  //专辑对应的主播名字 (nickName)
                    albumInfoIndex.setCreateTime(new Date());
                    countDownLatch.countDown(); //每执行完一个任务,就将countDownLatch减1,2
                } catch (Exception e) {
                    throw new RuntimeException(e);
                }
            }
        });

        executorService.execute(new Runnable() {
            @Override
            public void run() {
                //2.3 远程调用专辑微服务获取分类信息
                Result<BaseCategoryView> baseCategoryViewResult=albumInfoFeignClient.getAlbumInfoCatergoryView(albumId);
                BaseCategoryView categoryViewData = baseCategoryViewResult.getData();
                if (categoryViewData == null){
                    log.error("搜索微服务远程调用专辑微服务失败,获取分类信息失败");
                    throw new GuiguException(201,"搜索微服务远程调用专辑微服务失败,获取分类信息失败");
                }
                albumInfoIndex.setCategory1Id(categoryViewData.getCategory1Id());
                albumInfoIndex.setCategory2Id(categoryViewData.getCategory2Id());
                albumInfoIndex.setCategory3Id(categoryViewData.getCategory3Id());
                countDownLatch.countDown(); //每执行完一个任务,就将countDownLatch减1,1
            }
        });
        executorService.execute(new Runnable() {
            @Override
            public void run() {
                //2.4 远程调用专辑微服务获取四维信息
                Result<AlbumStatVo> albumStatVoResult=albumInfoFeignClient.getAlbumStatNum(albumId);
                AlbumStatVo albumStatVoData = albumStatVoResult.getData();
                if (albumStatVoData == null){
                    log.error("搜索微服务远程调用专辑微服务失败,获取四维信息失败");
                    throw new GuiguException(201,"搜索微服务远程调用专辑微服务失败,获取四维信息失败");
                }
                Integer playStatNum = albumStatVoData.getPlayStatNum();
                albumInfoIndex.setPlayStatNum(playStatNum);
                Integer subscribeStatNum = albumStatVoData.getSubscribeStatNum();
                albumInfoIndex.setSubscribeStatNum(subscribeStatNum);
                Integer buyStatNum = albumStatVoData.getBuyStatNum();
                albumInfoIndex.setBuyStatNum(buyStatNum);
                Integer commentStatNum = albumStatVoData.getCommentStatNum();
                albumInfoIndex.setCommentStatNum(commentStatNum);
                //Double hotScore = playStatNum * 0.6 + subscribeStatNum * 0.2 + buyStatNum * 0.1 + commentStatNum * 0.1;
                Double hotScore=new Random().nextDouble();
                albumInfoIndex.setHotScore(hotScore);
                countDownLatch.countDown(); //每执行完一个任务,就将countDownLatch减1,0
            }
        });

        countDownLatch.await(); //一旦coundownLatch减为0,则所有任务执行完毕,继续执行下面的代码
        //3.将文档对象放到albumInfoIndex索引库中
        AlbumInfoIndex saveFlag = albumInfoRepository.save(albumInfoIndex);
        long endTime = System.currentTimeMillis();
        log.info("保存专辑: {}到es,耗时:{}ms",albumId,(endTime-startTime));


        //将专辑信息保存到提示信息索引库中
        //创建提示词的文档对象
        SuggestIndex suggestIndex = new SuggestIndex();
        suggestIndex.setId(UUID.randomUUID().toString().replaceAll("-",""));
        suggestIndex.setTitle(albumInfoIndex.getAlbumTitle());
        suggestIndex.setKeyword(new Completion(new String[]{albumInfoIndex.getAlbumTitle()})); //专辑的内容
        suggestIndex.setKeywordPinyin(new Completion(new String[]{PinYinUtils.toHanyuPinyin(albumInfoIndex.getAlbumTitle())}));//专辑的拼音
        suggestIndex.setKeywordSequence(new Completion(new String[]{PinYinUtils.getFirstLetter(albumInfoIndex.getAlbumTitle())}));//专辑的首字母
        suggestIndexRepository.save(suggestIndex);

        return saveFlag != null;
    }

    @Override
    public boolean albumOffSale(Long albumId) {
        albumInfoRepository.deleteById(albumId);
        return true;
    }

    /**
     * 异步优化第三种方式
     * 异步编排
     * 函数式接口有很多:
     * Function<T,R>:   有参有返回值 ,   R apply(T t)      参数是T,返回值是R,
     * Supplier<T>:     无参入有返回值,  T get()           返回值是T
     * Consumer<T>:     有参无返回值,    void accept(T t)  参数是T
     *
     * @param albumId
     * @return
     */
    @Override
    @SneakyThrows
    public Map<String, Object> getAlbumDetail(Long albumId) {
        //Map中的key有4个 albumInfo:专辑信息  albumStatVo:辑统计信息 baseCategoryView:分类信息 announcer:主播信息

        //1.创建一个返回结果的对象
        HashMap<String, Object> resultMap = new HashMap<>();
        //2.给返回结果的对象赋值
        CompletableFuture<Long> completableFuture = CompletableFuture.supplyAsync(() -> {
            log.info("执行远程调用获取专辑的基本信息对应的线程:{}"+Thread.currentThread().getName());
            //2.1远程调用获取专辑的基本信息
            Result<AlbumInfo> albumInfoResult = albumInfoFeignClient.getAlbumInfoAndTag(albumId);
            AlbumInfo albumInfoData = albumInfoResult.getData();
            if (albumInfoData == null) {
                throw new GuiguException(201, "专辑不存在");
            }
            resultMap.put("albumInfo", albumInfoData);
            return albumInfoData.getUserId();
        },threadPoolExecutor);

        CompletableFuture<Void> userInfoFuture = completableFuture.thenAcceptAsync(userId -> {
            log.info("执行远程调用获取专辑的主播信息对应的线程:{}"+Thread.currentThread().getName());
            //2.4远程调用获取专辑的主播信息
            Result<UserInfoVo> userInfoVoResult = userInfoFeignClient.getUserInfoByUserId(userId);
            UserInfoVo userInfoVoData = userInfoVoResult.getData();
            if (userInfoVoData == null) {
                throw new GuiguException(201, "专辑不存在");
            }
            resultMap.put("announcer", userInfoVoData);
        },threadPoolExecutor);

        CompletableFuture<Void> AlbumStatVoFuture = CompletableFuture.runAsync(() -> {
            log.info("执行远程调用获取专辑的统计信息对应的线程:{}"+Thread.currentThread().getName());
            //2.2远程调用获取专辑的统计信息
            Result<AlbumStatVo> albumStatVoResult = albumInfoFeignClient.getAlbumStatNum(albumId);
            AlbumStatVo albumStatVoData = albumStatVoResult.getData();
            if (albumStatVoData == null) {
                throw new GuiguException(201, "专辑不存在");
            }
            resultMap.put("albumStatVo", albumStatVoData);
        },threadPoolExecutor);

        CompletableFuture<Void> BaseCategoryViewFuture = CompletableFuture.runAsync(() -> {
            log.info("执行远程调用获取专辑的分类信息对应的线程:{}"+Thread.currentThread().getName());
            //2.3远程调用获取专辑的分类信息
            Result<BaseCategoryView> baseCategoryViewResult = albumInfoFeignClient.getAlbumInfoCatergoryView(albumId);
            BaseCategoryView baseCategoryViewData = baseCategoryViewResult.getData();
            if (baseCategoryViewData == null) {
                throw new GuiguException(201, "专辑不存在");
            }
            resultMap.put("baseCategoryView", baseCategoryViewData);

        },threadPoolExecutor);
        //等四个线程都执行完毕,在执行下面的任务
        CompletableFuture.allOf(userInfoFuture, AlbumStatVoFuture, BaseCategoryViewFuture).get();
        log.info("执行返回结果任务对应的线程:{}"+Thread.currentThread().getName());
        //3.返回结果对象
        return resultMap;
    }

    @Override
    @SneakyThrows
    public Boolean rankPreHeat() {
        //1.查询所有的一级分类
        Result<List<BaseCategory1>>  allCategory1= albumInfoFeignClient.getAllCategory1();
        List<BaseCategory1> allCategory1Data = allCategory1.getData();
        if (CollectionUtils.isEmpty(allCategory1Data)){
            throw new GuiguException(201,"该平台不存在一级分类列表");
        }
        //遍历所有的一级分类列表
        for (BaseCategory1 baseCategory1 : allCategory1Data) {
            //2.1 查询每一个一级分类下的5个维度的专辑留下10个最高的
            //创建一个String类型的数组
            String[] dimensions = new String[]{"hotScore","playStatNum","subscribeStatNum","buyStatNum","commentStatNum"};
            //查询所有维度(从es中查询)
            for (String dimension : dimensions) {
                SearchResponse<AlbumInfoIndex> searchResponse = elasticsearchClient.search(srb -> srb
                        .index("albuminfo")
                        .query(qb -> qb
                                .term(tqb -> tqb
                                        .field("category1Id")
                                        .value(baseCategory1.getId().toString())))
                        .sort(sob -> sob
                                .field(fqb -> fqb
                                        .field(dimension)
                                        .order(SortOrder.Desc)))
                        .size(10), AlbumInfoIndex.class);

                //解析数据
                List<AlbumInfoIndex> albumInfoIndexList = searchResponse.hits().hits().stream().map(hit -> {
                    AlbumInfoIndex source = hit.source();
                    return source;
                }).collect(Collectors.toList());

                //2.2存到Redis中(1.String 2.Set 3.Zset 4.List 5.Hash(大key,小key)
                //json格式的字符串数组["{"name":"张三","age":18}","{"name":"李四","age":19}"]
                redisTemplate.opsForHash().put(baseCategory1.getId().toString(),dimension, JSONObject.toJSONString(albumInfoIndexList));
            }
        }
        return true;
    }

    /**
     * 查询排行榜信息
     * @param c1Id
     * @param dimension
     * @return
     */
    @Override
    public List<AlbumInfoIndexVo> findRankingList(Long c1Id, String dimension) {
        String result = (String) redisTemplate.opsForHash().get(c1Id.toString(), dimension);
        //反序列化
        List<AlbumInfoIndexVo> albumInfoIndexVoList = JSONObject.parseArray(result, AlbumInfoIndexVo.class);
        return albumInfoIndexVoList;
    }

    @Override
    public void batchAlbumOffSale() {
        albumInfoRepository.deleteAll();
    }
}
